package com.credithc.signature.job.configure.rabbitmq;

import com.credithc.signature.job.mq.JobConsumer;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

/**
 * 描述：rabbitmq配置<br/>
 * 作者：hujingbo
 * 创建：2019/6/4 14:40
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Configuration
@Data
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualHost;

    @Value("${spring.rabbitmq.customConfig.exchangeName:signature-job-exchange}")
    private String exchangeName;

    @Value("${spring.rabbitmq.customConfig.queueName:signature-job-queue}")
    private String queueName;

    @Value("${spring.rabbitmq.cache.connection.size:30}")
    private int connectionCacheSize;

    @Value("${spring.rabbitmq.cache.channel.size:25}")
    private int channelCacheSize;

    @Value("${spring.rabbitmq.customConfig.maxConcurrentConsumers:1}")
    private int maxConcurrentConsumers;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirm;

    @Value("${spring.rabbitmq.callback.channel.prefetchCount:10}")
    private int prefetchCount;

    @Value("${spring.rabbitmq.customConfig.canal.exchangeName:signature-canal-exchange}")
    private String canalExchangeName;

    @Value("${spring.rabbitmq.customConfig.canal.queueName:signature-canal-queue}")
    private String canalQueueName;

    @Value("${spring.rabbitmq.customConfig.canal.routingKey:canal-exchange-routing}")
    private String canalRoutingKey;


    /**
     * 连接工厂
     *
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        connectionFactory.setChannelCacheSize(channelCacheSize);
        connectionFactory.setConnectionCacheSize(connectionCacheSize);
        connectionFactory.setPublisherConfirms(publisherConfirm);
        connectionFactory.setConnectionLimit(connectionCacheSize);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public Queue queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(queueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    @Bean
    public DirectExchange exchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }

    @Bean
    public Binding binding(RabbitAdmin rabbitAdmin) {
        Queue queue = queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(exchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

//    @Bean
//    public Queue canalQueue(RabbitAdmin rabbitAdmin) {
//        Queue queue = new Queue(canalQueueName, true, false, false);
//        rabbitAdmin.declareQueue(queue);
//        return queue;
//    }
//
//    @Bean
//    public DirectExchange canalExchange(RabbitAdmin rabbitAdmin) {
//        DirectExchange directExchange = new DirectExchange(canalExchangeName, true, false);
//        rabbitAdmin.declareExchange(directExchange);
//        return directExchange;
//    }
//
//    @Bean
//    public Binding canalBinding(RabbitAdmin rabbitAdmin) {
//        Queue queue = canalQueue(rabbitAdmin);
//        Binding binding = BindingBuilder.bind(queue).to(canalExchange(rabbitAdmin)).with(canalRoutingKey);
//        rabbitAdmin.declareBinding(binding);
//        return binding;
//    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(JobConsumer jobConsumer) {
        return getSimpleMessageListenerContainer(jobConsumer, queue(rabbitAdmin()));
    }

//    @Bean
//    public SimpleMessageListenerContainer messageCanalContainer(CanalMqConsumer canalMqConsumer) {
//        return getSimpleMessageListenerContainer(canalMqConsumer, canalQueue(rabbitAdmin()));
//    }

    private SimpleMessageListenerContainer getSimpleMessageListenerContainer(ChannelAwareMessageListener listener, Queue queue) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueueNames(queue.getName());
        // 需要将channel暴露给listener才能手动确认，AcknowledgeMode.MANUAL时必须为ture
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(maxConcurrentConsumers);
        container.setConcurrentConsumers(maxConcurrentConsumers);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(listener);
        container.setPrefetchCount(prefetchCount);
        // 设为true，扔回queue头部，设为false，丢弃
        container.setDefaultRequeueRejected(true);
        container.setMessageConverter(jsonMessageConverter());
        return container;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMandatory(true);
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        return template;
    }
}
